package start.spring.basic.job.job;

import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.jboss.logging.MDC;
import org.quartz.JobExecutionContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;

import lombok.extern.slf4j.Slf4j;
import start.spring.basic.config.CustomUidGenerator;
import start.spring.basic.dynamic.DynamicJdbc;
import start.spring.basic.dynamic.DynamicRedisDAO;
import start.spring.basic.job.job.service.IJobService;
import start.spring.basic.job.quartz.QuartzConstant;
import start.spring.basic.log.LogConstant;
import start.spring.basic.util.common.StringUtil;
import start.spring.basic.util.thread.NamedThreadFactory;

/**
 * ****************************************************************************
 * 自动任务创建基类，提供自动任务的公共方法
 *
 * @author(作者)：xuyongyun	
 * @date(创建日期)：2018年8月29日
 ******************************************************************************
 */
@Slf4j
public class BaseJob {
	
	@Autowired
	private DiscoveryClient client;
	
	@Autowired
	protected DynamicJdbc dao;
	
	@Autowired
	protected DynamicRedisDAO redis;
	
	@Autowired
	protected CustomUidGenerator idGenerator;
	
	@Autowired
	IJobService jobService;
	
	@Value("${spring.application.name:job}")
    private String serviceId;

	/**
	 * job是否可执行
	 *	
	 * @param className job的执行类名
	 * @param context 任务配置信息
	 * @return Boolean.TRUE
	 * @since           1.0
	 */
	@SuppressWarnings("rawtypes")
	public boolean jobCanExcute(String className, JobExecutionContext context) {
		
		String traceId = StringUtil.getUuid();
		MDC.put(LogConstant.REQUEST_KEY_TRACEID, traceId);
		
		//默认可执行
		boolean canExcte = true;
		
		try {
			AtomicInteger runThreadNum = jobService.getRunJobCounter();
			//获取自动任务服务总数
			List<ServiceInstance> jobList = client.getInstances(serviceId);
			log.debug("{}的服务数量：{}", serviceId, jobList.size());
			if(jobList!=null&&jobList.size()>1) {
				//执行自动任务的服务总数
				int serviceCount = jobList.size();
				//查询有效任务总数
				String sql = "select * from GG_PZ_JOB t where yxbz = 'Y' ";
				List resultList = dao.query(null, sql, null);
				if(resultList!=null&&!resultList.isEmpty()) {
					//当前存在有效任务
					int taskCount = resultList.size();
					
					//平均每个服务需要执行的任务数量
					int avgCount = (int)Math.ceil(taskCount*1.0/serviceCount);
					if(runThreadNum.get()>=avgCount) {
						//当前受管并发执行任务数已达到平均值，暂时不在执行新任务
						canExcte = false;
						log.info("当前受管并发执行任务数已达到平均值，暂时不再执行新任务！");
					}
					
				}else {
					//当前所有任务均无效
					canExcte = false;
					log.debug("{}已不是有效任务，不发起调度！",className);
				}
			}
			
			if(canExcte) {
				//查询是否有效任务
				String sql = "select t.status,case t.status when '1' then timestampdiff(MINUTE,t.kssj,date_format(now(), '%Y-%m-%d %H:%i:%s')) else 0 end sjc,timeout cssj,threadnum from GG_PZ_JOB t where t.jobclass = ? and yxbz = 'Y' ";
				String[] prams = new String[] {className};
				List resultList = dao.query(null, sql, prams);
				if(resultList!=null&&resultList.size()>0) {
					//获取到锁
					Map tmpMap = (Map) resultList.get(0);
					String status = String.valueOf(tmpMap.get("STATUS"));
					//配置的超时时间，单位分钟
					long timeOut = Long.parseLong(StringUtil.trimObject(tmpMap.get("CSSJ"),"1"));
					//执行线程数
					String threadNum = StringUtil.trimObject(tmpMap.get("THREADNUM"),"1");
					
					//防止任务被多个受管并发执行，先获取redis锁
					boolean lock = redis.setnx(null, className, String.valueOf(System.currentTimeMillis()));
					if(!lock) {
						//未获取到锁，判断锁是否已超时
						String lockTime = redis.get(null, className);
						if(StringUtil.isEmpty(lockTime)) {
							//redis存储值有问题，重新锁定
							lock = true;
							redis.set(null, className, String.valueOf(System.currentTimeMillis()));
						}else {
							//redis值正常，判断锁是否已超时了(由于上次解锁失败导致本次没获取到锁)
							long timeSub = System.currentTimeMillis() - Long.parseLong(lockTime);
							//锁已超时
							if(timeSub >=  timeOut*60*1000) {
								lock = true;
								redis.set(null, className, String.valueOf(System.currentTimeMillis()));
							}
						}
					}
					
					if(lock) {
						runThreadNum.getAndIncrement();
						log.info("{}开始运行，当前运行任务数：{}", className, runThreadNum.get());
						//任务状态运行中
						if(QuartzConstant.JOB_STATUS_RUNNING.equals(status)) {
							String sjcxs = String.valueOf(tmpMap.get("SJC"));
							if(StringUtil.isEmpty(sjcxs)) {
								canExcte = false;
								log.warn("kssj为空，不能判断任务当前的执行情况，需要人工干预，不发起调度！");
								
								//设置任务执行失败
								String jobStatus = QuartzConstant.JOB_STATUS_FAIL;
								String errMsg = "任务开始时间为空，无法判断是否执行超时，请手工重置任务状态！";
								//记录执行失败日志
								reJobEnd(jobStatus, errMsg, className);
							}else {
								double sjc = Double.valueOf(sjcxs).doubleValue();
								
								//job运行开始时间距离当前时间小于CANEXCUTE_SJC对应的值，认为job正在运行
								if(sjc<timeOut) {
									canExcte = false;
									log.warn("{}执行中，本次不执行！", className);
									
									//设置任务执行失败
									String jobStatus = QuartzConstant.JOB_STATUS_FAIL;
									String errMsg = "任务执行中，本次不发起调度！";
									//记录执行失败日志
									reJobEnd(jobStatus, errMsg, className);
								}
							}
						}
						if(canExcte) {
							context.getJobDetail().getJobDataMap().put(QuartzConstant.THREAD_NUM, threadNum);
							//任务可以执行，则设置锁的超时时间为任务超时时间
							redis.expire(null, className, timeOut*60);
						}else {
							//任务不可以执行，直接解锁
							redis.expire(null, className, 1);
						}
					}
				}else {
					canExcte = false;
					log.debug("{}已不是有效任务，不发起调度！",className);
				}
			}
		}catch(Exception e) {
			canExcte = false;
			log.error("查询运行状态出错，不能判断任务当前的执行情况，不发起调度！",e);
		}
		return canExcte;
	}
	
	/**
	 * 记录任务开始执行
	 *
	 * @param className 执行类名
	 * @since           1.0
	 */
	public void reJobStart(String className) {
		jobService.reJobStart(className);
	}
	
	/**
	 * 记录任务结束执行状态
	 *
	 * @param status 任务状态
	 * @param errMsg 错误消息
	 * @param className 执行类名
	 * @since           1.0
	 */
	public void reJobEnd(String status,String errMsg,String className) {
		jobService.reJobEnd(status,errMsg,className,null);
		//清理redis锁，使任务可以重新发起
		redis.expire(null, className, 1);
		MDC.clear();
	}
	
	/**
	 * 获取多线程job的线程池
	 *	
	 * @param threadName
	 * @param threadNum
	 * @return
	 * @since           1.0
	 */
	protected ThreadPoolExecutor getThreadPool(String threadName, int threadNum) {
		NamedThreadFactory threadFactory = new NamedThreadFactory(threadName);
		ThreadPoolExecutor cachedThreadPool = new ThreadPoolExecutor(8, 128, 0, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(1024), threadFactory, new ThreadPoolExecutor.AbortPolicy());
		return cachedThreadPool;
	}
	
	/**
	 * 获取多线程job的超时时间
	 *	
	 * @param context
	 * @return
	 * @since           1.0
	 */
	protected int getTimeOut(JobExecutionContext context) {
		return Integer.parseInt(
				StringUtil.trimObject(context.getJobDetail().getJobDataMap().get(QuartzConstant.TIME_OUT), "30"));
	}
	
	/**
	 * 获取多线程任务的并发数量
	 *	
	 * @param context
	 * @return
	 * @since           1.0
	 */
	protected int getThreadNum(JobExecutionContext context) {
		return Integer.parseInt(
				StringUtil.trimObject(context.getJobDetail().getJobDataMap().get(QuartzConstant.THREAD_NUM), "1"));
	}
}
